package boot.spring.scheduler;

import boot.spring.model.entity.Track;
import boot.spring.model.po.TrackMessage;
import boot.spring.service.TrackService;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.List;

@Slf4j
@Component
public class TrackSimulatorScheduler {

    private static final Duration OFFSET =
            Duration.between(LocalDateTime.now(), LocalDateTime.of(2017, 9, 7, 0, 0));

    @Resource
    private TrackService trackService;

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @Resource
    private ObjectMapper mapper;

//    @Scheduled(fixedRate = 1000)
    public void sendTracksToKafka() {
        // 获取当前时间
        LocalDateTime now = LocalDateTime.now();

        // 提取时分秒
        int hour = now.getHour();
        int minute = now.getMinute();
        int second = now.getSecond();

        // 构建2017年9月7日相同时分秒
        LocalDateTime startTime = LocalDateTime.of(2017, 9, 7, hour, minute, second);
        LocalDateTime endTime = startTime.plusSeconds(1);
        log.debug("Sending tracks to simulator at {} to {}", startTime, endTime);

        List<Track> tracks = trackService.selectByTs(startTime, endTime);

        for (Track track : tracks) {
            try {
                LocalDateTime dt = LocalDateTime.of(now.getYear(), now.getMonth(), now.getDayOfMonth(),
                        track.getTs().getHour(), track.getTs().getMinute(), track.getTs().getSecond());
                long sTm = dt.atZone(ZoneId.systemDefault())
                        .toInstant()
                        .toEpochMilli();
                TrackMessage trackMessage = getTrackMessage(track, sTm);
                String json = mapper.writeValueAsString(trackMessage);
                kafkaTemplate.send("track", track.getDeviceId().toString(), json);
                log.debug("发送轨迹: {}", json);
            } catch (Exception e) {
                log.error("轨迹序列化或发送失败", e);
            }
        }
    }

    private static TrackMessage getTrackMessage(Track track, long sTm) {
        TrackMessage trackMessage = new TrackMessage();
        trackMessage.setSTm(sTm);
        trackMessage.setGTm(sTm);
        trackMessage.setDevId(track.getDeviceId());
        trackMessage.setLon(track.getLon());
        trackMessage.setLat(track.getLat());
        trackMessage.setAlt(track.getAlt());
        trackMessage.setV(track.getSpeed());
        trackMessage.setDir(track.getDir());
        trackMessage.setCType(1);
        trackMessage.setVer(100);
        trackMessage.setSource(1);
        return trackMessage;
    }
}
